package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

/**
    TODO  算子--patititonBy 重新分区
      1、PartitionBy的作用是将数据进行重分区,之前学习的扩大和缩减分区只是为了改变分区的数量，数据和分区关系不考虑
      2、ParititonBy算子的主要目的是将数据改变分区。
      3、需要传递一个（分区器）分区规则
      4、Spark在当前场合下默认提供了3个分区器，一般常用的是2个：
          1）RangePartitioner:范围分区器（Hbase预分区），排序要求数据必须可以比较
          2）HashPartitioner:哈希分区器，默认Spark提供的分区器
      5、数据格式中KV类型的数据处理比较常见，所以Spark的RDD 专门提供了一些方法（算子）来处理KV类型的
      6、一个对象能用什么方法取决于什么？类型和泛型无关。
      7、编译出现错误时，编译器会尝试在作用域中查找相应的转换规则，进行二次编译（隐式转换）
      8、所以KV类型的数据的操作方法，在RDD中都不存在，KV操作都是在PairRDDFunctions类中完成的
 */

object Spark_RDD_Pper_Transform_PartitionBy {
  def main(args: Array[String]): Unit = {


  //TODO 1、获得连接
  val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local[*]")
  val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑 --测试分区
  val rdd= sc.makeRDD(List(("a",1),"b" -> 2,("c" -> 3),"d" -> 4),2)

    //3.1 测试初始设置分区数
    rdd.saveAsTextFile("output")
    //3.2 默认是HashPartitioner分区,就直接将分区数传进去,这个按照哈希散列的方式将数据打散

    val newRdd2: RDD[(String, Int)] = rdd.partitionBy(new HashPartitioner(4))

    //测试修改后的分区数
    newRdd2.saveAsTextFile("output1")
  //TODO 2、关闭连接
  sc.stop()
}
}